package edu.hit.crawler.util;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

public class FileSystemUtil {
	
	public static final int Total = 0, Part = 1;
		
	public static class FileItem implements Comparable<FileItem>{
		private FileStatus fStatus;
		private float fileSize;
		
		public FileItem() {
			
		}
		
		public FileItem(FileStatus f, float fileSize) {
			this.setfStatus(f);
			this.setFileSize(fileSize);
		}
		
		@Override
		public int compareTo(FileItem fi) {
			return (int)(this.getFileSize() - fi.getFileSize());
		}

		public void setfStatus(FileStatus fStatus) {
			this.fStatus = fStatus;
		}

		public FileStatus getfStatus() {
			return fStatus;
		}

		public void setFileSize(float fileSize) {
			this.fileSize = fileSize;
		}

		public float getFileSize() {
			return fileSize;
		}	
	}

	public static void CopyFilesOnHDFS(FileSystem fs, Configuration conf, Path src, Path dest, boolean deleteSrc) throws IOException {
		
		try {
			if(fs.exists(dest)) {
				fs.delete(dest, true);
			}			
			fs.mkdirs(dest);
			
			if (fs.isFile(src)) {
				FileUtil.copy(fs, src, fs, dest, false, conf);
			} else {
				FileStatus[] fStatus = fs.listStatus(src);
				
				if(fStatus.length != 0) {
					for(FileStatus f : fStatus) {
						
						FileUtil.copy(fs, f.getPath(), fs, dest, false, conf);
					}
				}
			}
		} catch (IOException ex) {
			System.err.println("copy error:" + ex.getMessage());
		}
	} 
	
	/**
	 * 在HDFS文件系统上复制文件
	 * @param fs FileSystem 
	 * @param conf Configuration
	 * @param src 源文件或源文件目录
	 * @param dest 目标文件目录
	 * @param deleteSrc 是否删除源文件或源文件目录
	 * @param limitSize 限制的复制文件总的大小（MB）
	 * @param copyType 复制文件类型（全部复制、部分复制）
	 * @throws IOException
	 */
	/*public static void CopyFilesOnHDFS(FileSystem fs, Configuration conf, Path src, Path dest, boolean deleteSrc, float limitSize, int copyType) throws IOException {
		try {
			//if(fs.exists(dest)) {
				//fs.delete(dest, true);
			//}			
			//fs.mkdirs(dest);
			
			if(!fs.exists(dest)) {
				fs.mkdirs(dest);
			}
			
			if (fs.isFile(src)) {
				FileUtil.copy(fs, src, fs, dest, deleteSrc, conf);
			} else {
				FileStatus[] fStatus = fs.listStatus(src);
				
				if(fStatus.length != 0) {
					List<FileItem> fItems = new ArrayList<FileItem>();
					
					for(FileStatus f : fStatus) {
						
						fItems.add(new FileItem(f, sizeOfFile_MB(f)));
					}
					
					Collections.sort(fItems);
					
					if (copyType == Total) {
						for (FileItem fi : fItems) {

							FileUtil.copy(fs, fi.getfStatus().getPath(), fs,
									dest, deleteSrc, conf);
						}
					} else {
						float count = 0f;

						for (FileItem fi : fItems) {
							if (count < limitSize) {
								FileUtil.copy(fs, fi.getfStatus().getPath(),
										fs, dest, deleteSrc, conf);
								count += fi.getFileSize();
							} else {
								break;
							}
						}
					}
				}
			}
		} catch (IOException ex) {
			System.err.println("copy error:" + ex.getMessage());
		}
	}*/ 
	
	/**
	 * 在HDFS文件系统上复制文件
	 * @param fs FileSystem 
	 * @param conf Configuration
	 * @param src 源文件或源文件目录
	 * @param dest 目标文件目录
	 * @param deleteSrc 是否删除源文件或源文件目录
	 * @param limitCount 限制的复制文件总的个数
	 * @throws IOException
	 */
	public static void CopyFilesOnHDFS(FileSystem fs, Configuration conf, Path src, Path dest, boolean deleteSrc, int limitCount) throws IOException {
		try {
			
			if(!fs.exists(dest)) {
				fs.mkdirs(dest);
			}
			
			if (limitCount >= 0) {

				dest = new Path(dest, src.getName());
				
				FileStatus[] fStatus = fs.listStatus(src);

				if (fStatus.length != 0) {

					for (int i = 0; (i < limitCount && i < fStatus.length); i++) {
						FileUtil.copy(fs, fStatus[i].getPath(), fs, new Path(dest, fStatus[i].getPath().getName()), deleteSrc, conf);
					}
				}
			}
			
		} catch (IOException ex) {
			System.err.println("copy error:" + ex.getMessage());
		}
	} 
	
	public static float sizeOfFile_MB(FileStatus fileStatus) {
		float res = 0f;
		
		if (fileStatus != null) {
			
			long len = fileStatus.getLen();  //单位为Byte
			
			res = len / 1024f / 1024f ;    //单位为MB
		}
		return res;
	}
	
	/**
	 * 将HDFS文件系统上文件复制到本地
	 * @param fs FileSystem 
	 * @param conf Configuration
	 * @param src 源文件或源文件目录
	 * @param dest 目标文件目录
	 * @throws IOException
	 */
	public static void CopyFilesToLocal(FileSystem fs, Configuration conf, Path src, String dest) throws IOException {
		try {

			File destFile = new File(dest);

			if (destFile.exists()) {
				destFile.delete();
			}

			destFile = new File(dest);
			destFile.mkdirs();

			FileStatus[] fStatus = fs.listStatus(src);

			if (fStatus.length != 0) {

				for (int i = 0; i < fStatus.length; i++) {

					FileStatus[] fStatusF = fs.listStatus(fStatus[i].getPath());

					if (fStatusF.length != 0) {

						for (int j = 0; j < fStatusF.length; j++) {
							String dFile = dest + System.getProperty("file.separator") + fStatus[i].getPath().getName() 
							+ System.getProperty("file.separator") + fStatusF[j].getPath().getName();
							FileUtil.copyToLocal(fs, fStatusF[j].getPath(),	dFile, conf);
						}
					}
				}
			}
		} catch (IOException ex) {
			System.err.println("copyToLocal error:" + ex.getMessage());
		}
	} 
	
	/**
	 * 将本地文件复制到HDFS文件系统上
	 * @param fs FileSystem 
	 * @param conf Configuration
	 * @param src 源文件或源文件目录
	 * @param dest 目标文件目录
	 * @throws IOException
	 */
	public static void CopyFilesFromLocal(FileSystem fs, Configuration conf, String src, Path dest) throws IOException {
		try {

			if (fs.exists(dest)) {
				fs.delete(dest, true);
			}
			
			File srcFile = new File(src);
			
			if (srcFile.isDirectory()) {
				
				String[] files = srcFile.list();

				if (files.length != 0) {
										
					for (int i = 0; i < files.length; i++) {
						Path mdest = new Path(dest, files[i]);

						File mFile = new File(srcFile + System.getProperty("file.separator") + files[i]);
						
						if (mFile.isDirectory()) {
							String[] filesF = mFile.list();

							if (filesF.length != 0) {
								for (int j = 0; j < filesF.length; j++) {
									Path lPath = new Path(mdest, filesF[j]);
									String srcPath = mFile.getAbsolutePath() + System.getProperty("file.separator") + filesF[j];
									
									if(!fs.exists(lPath.getParent())) {
										fs.mkdirs(lPath.getParent());
									}
									
									FileUtil.copyFromLocal(srcPath, fs, lPath, conf);
								}
							}
						}
					}
				}
			}
		} catch (IOException ex) {
			System.err.println("copyToLocal error:" + ex.getMessage());
		}
	} 
	
	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(conf);
		
		Path pSrc = new Path("crawler/out/");
		Path pSrc1 = new Path("crawler/in/");
		Path pSrc2 = new Path("crawler/format/");
		Path pSrc3 = new Path("crawler/test/");
		
		//FileSystemUtil.CopyFilesOnHDFS(fs, conf, pSrc, pDest, false);
		
		//FileSystemUtil.CopyFilesOnHDFS(fs, conf, pSrc, pDest, false, 0.000039f);
		//FileSystemUtil.CopyFilesOnHDFS(fs, conf, pSrc, pDest, false, 2f, FileSystemUtil.Part);
		
		//FileSystemUtil.CopyFilesToLocal(fs, conf, pSrc, "C:\\out");
		//FileSystemUtil.CopyFilesToLocal(fs, conf, pSrc1, "C:\\in");
		//FileSystemUtil.CopyFilesToLocal(fs, conf, pSrc2, "C:\\format");
		FileSystemUtil.CopyFilesFromLocal(fs, conf, "C:\\format", pSrc3);
		
		System.out.println("Done.");
	}
	
	public static void DeleteAllFiles(String fileName) {
		File file = new File(fileName);
		
		if(file != null && file.exists()) {
			DeleteAllFiles(file);
		}
	}
	
	public static void DeleteAllFiles(File file) {	
		if(file.isFile()) {
			file.delete();
		} else if(file.isDirectory()) {
			
			File[] files = file.listFiles(); 
			
			if(files.length != 0) {
				for(int i = 0; i < files.length; i++) {
					DeleteAllFiles(files[i]);
				}
			}	
			
			file.delete();
		}
	}
}

class FileUtil {
	public static boolean copy(FileSystem srcFS, Path src, FileSystem dstFS,
			Path dst, boolean deleteSource, Configuration conf)
			throws IOException {
		
		if (srcFS.isFile(src)) {
			InputStream in = null;
			OutputStream out = null;
			try {
				in = srcFS.open(src);
				out = dstFS.create(dst);
				IOUtils.copyBytes(in, out, conf, true);
			} catch (IOException e) {
				IOUtils.closeStream(out);
				IOUtils.closeStream(in);
				throw e;
			}

		} else if (srcFS.getFileStatus(src).isDir()) {
						
		} else {
			throw new IOException((new StringBuilder()).append(src.toString()).append(": No such file").toString());
		}
		if (deleteSource)
			return srcFS.delete(src, true);
		else
			return true;
	}	
	
	public static boolean copyToLocal(FileSystem srcFS, Path src, String dstFS,
			Configuration conf)
			throws IOException {
				
		File destFile = new File(dstFS);
		if(destFile.exists()) {
			destFile.delete();
		}
		
		destFile = new File(dstFS);
		File parent = new File(destFile.getParent()); 
		if(!parent.exists()) {
			parent.mkdirs();
		}
		
		if (srcFS.isFile(src)) {
			InputStream in = null;
			OutputStream out = null;
			try {
				in = srcFS.open(src);
				out = new FileOutputStream(destFile);
				IOUtils.copyBytes(in, out, conf, true);
			} catch (IOException e) {
				IOUtils.closeStream(out);
				IOUtils.closeStream(in);
				throw e;
			}

		} else if (srcFS.getFileStatus(src).isDir()) {
				System.out.println("is directory.");		
		} else {
			throw new IOException((new StringBuilder()).append(src.toString()).append(": No such file").toString());
		}
		
		return true;
	}	
	
	public static boolean copyFromLocal(String srcFS, FileSystem dstFS,
			Path dst, Configuration conf)
			throws IOException {
		
		File srcFile = new File(srcFS);
		if(dstFS.exists(dst)) {
			dstFS.delete(dst, true);
		}		
		
		if (srcFile.isFile()) {
			InputStream in = null;
			OutputStream out = null;
			try {
				in = new FileInputStream(srcFS);
				out = dstFS.create(dst);
				IOUtils.copyBytes(in, out, conf, true);
			} catch (IOException e) {
				IOUtils.closeStream(out);
				IOUtils.closeStream(in);
				throw e;
			}

		}  else {
			throw new IOException((new StringBuilder()).append(srcFS).append(": No such file").toString());
		}

		return true;
	}	
}
